﻿package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"runtime"
	"simudata/mqttclient"
	"strconv"
	"syscall"
	"time"

	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
	"github.com/influxdata/influxdb-client-go/v2/api"
	http2 "github.com/influxdata/influxdb-client-go/v2/api/http"
	"github.com/sirupsen/logrus"
	"github.com/spf13/viper"
	"golang.org/x/sync/errgroup"
	"google.golang.org/protobuf/proto"
	"gopkg.in/natefinch/lumberjack.v2"
)

// influxdb数据库客户端对象
var storageClient influxdb2.Client

// influxdb客户端数据写入接口，非阻塞模式
var storageWriteAPI api.WriteAPI

// mqtt客户端对象
var mqClient *mqttclient.MqttClient

func initLogger() {
	enable := viper.GetBool(logFileEnableKey)
	if enable {
		name := viper.GetString(logFileNameKey)
		maxSize := viper.GetInt(logFileMaxSizeKey)
		maxBackups := viper.GetInt(logFileMaxBackupSizeKey)
		maxAge := viper.GetInt(logFileMaxAgeKey)
		compress := viper.GetBool(logFileCompressKey)
		logFile := &lumberjack.Logger{
			Filename:   name,
			MaxSize:    maxSize,
			MaxBackups: maxBackups,
			MaxAge:     maxAge,
			Compress:   compress,
		}
		logrus.SetOutput(logFile)
	}
	logrus.SetFormatter(&logrus.TextFormatter{TimestampFormat: "2006-01-02 15:04:05.999"})
}

// 初始化配置参数，设置默认参数和从配置文件中读取覆盖
func initConfig() {
	viper.SetConfigName("config")
	viper.SetConfigType("yaml")
	viper.SetConfigFile("./config.yaml")
	viper.AddConfigPath(".")
	// 设置设备参数
	viper.SetDefault(deviceNameKey, randomString(6))
	viper.SetDefault(deviceSensorCount, 10)
	viper.SetDefault(deviceSampleRateKey, 100)
	viper.SetDefault(deviceStorageEnableKey, true)

	// 设置MQTT参数
	viper.SetDefault(mqttServerIPKey, "localhost")
	viper.SetDefault(mqttServerPortKey, 1883)
	viper.SetDefault(mqttWillEnableKey, false)
	viper.SetDefault(mqttQosKey, 0)
	viper.SetDefault(mqttRetainedKey, false)
	viper.SetDefault(mqttPublishChannelKey, "simu_device")

	// 设置定时器每秒的触发次数
	viper.SetDefault(tickerTriggersKey, 50)

	// 设置随机数生成区间
	viper.SetDefault(floatMaxValueKey, 100.0)
	viper.SetDefault(floatMinValueKey, 0.0)
	viper.SetDefault(floatPercisionKey, 6)

	// 设置influxdb的参数
	viper.SetDefault(influxdbOrgKey, "test")
	viper.SetDefault(influxdbBucketKey, "device")
	viper.SetDefault(influxdbTokenKey, "iWSIQa0AAnFwgJdAZd3sve2yupSTH5DECbN5iTKfQ5Oxs0fK84Pin1t7smHgen7hzyn_lPwdg5WVQc4Nl8OPow==")
	viper.SetDefault(influxdbBatchSizeKey, 50)
	viper.SetDefault(influxdbServerIPKey, "localhost")
	viper.SetDefault(influxdbServerPortKey, 8086)

	// 设置日志参数
	viper.SetDefault(logFileEnableKey, true)
	viper.SetDefault(logFileNameKey, "simudata.log")
	// 设置日志文件大小，单位M
	viper.SetDefault(logFileMaxSizeKey, 5)
	viper.SetDefault(logFileCompressKey, false)
	// 日志文件保留时长,单位天
	viper.SetDefault(logFileMaxAgeKey, 10)
	// 日志文件最多保留的文件数
	viper.SetDefault(logFileMaxBackupSizeKey, 20)

	// 读取配置文件参数
	viper.ReadInConfig()

	initLogger()

	viper.WriteConfig()

}

// influxdb数据写入失败时的回调
func onInfluxWriteFailed(batch string, httpError http2.Error, retryAttempts uint) bool {
	if retryAttempts > 3 {
		logrus.Error("因为", httpError.Error(), "导致influxdb写入数据超出尝试次数，数据将被丢弃:")
		logrus.Error(batch)
		return false
	}
	logrus.Warn("influxdb数据写入失败，原因:", httpError.Error(), " 即将进行写入重试")
	return true
}

// 生成模拟数据并发送到MQTT和写入influxdb
func generateAndSendSensorDatas(ctx context.Context, clock <-chan time.Time) error {
	publishChannelName := viper.GetString(mqttPublishChannelKey)
	deviceName := viper.GetString(deviceNameKey)
	deviceSensorCount := viper.GetInt(deviceSensorCount)
	deviceSampleRate := viper.GetInt64(deviceSampleRateKey)
	maxValue := viper.GetFloat64(floatMaxValueKey)
	minValue := viper.GetFloat64(floatMinValueKey)
	tickerTriggers := viper.GetInt64(tickerTriggersKey)
	interval := int64(time.Duration(1000 / tickerTriggers))
	sampleSize := int64(deviceSampleRate / tickerTriggers)
	if sampleSize == 0 {
		sampleSize = 1
	}
	timeStamp := int64(0)
	for {

		select {
		case _, ok := <-clock:
			if !ok {
				return errors.New("ticket closed")
			} else {
				if timeStamp == 0 {
					timeStamp = time.Now().UnixMilli()
				}
				message := generateSensorDatas(timeStamp, deviceName, deviceSensorCount, int64(deviceSampleRate/tickerTriggers), interval, minValue, maxValue)
				payload, err := proto.Marshal(message)
				if nil == err {
					err := mqClient.Publish(publishChannelName, payload)
					if nil == err && nil != storageWriteAPI {
						for _, data := range message.Datas {
							pt := influxdb2.NewPointWithMeasurement(deviceName).AddTag("sensor", strconv.FormatInt(int64(data.Id), 10)).AddField("value", data.Value).SetTime(time.UnixMilli(data.Timestamp))
							storageWriteAPI.WritePoint(pt)
						}

					}
				}
				timeStamp += interval
			}

		case <-ctx.Done():
			if storageWriteAPI != nil {
				storageWriteAPI.Flush()
			}
			logrus.Info("可能是收到了系统中断信号导致程序中断:", ctx.Err())
			return ctx.Err()
		}
	}
}

func main() {
	// 初始化配置
	initConfig()
	signalCtx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, os.Interrupt)
	defer stop()

	groupFunc, groupCtx := errgroup.WithContext(signalCtx)

	if runtime.GOOS == "darwin" {
		viper.Set(influxdbTokenKey, "0Ybba9a9eOVU9RxhUHS1ENGVnQ7olOthcXeEMczFh8mtAn008Qfy_OYqvJYxqhrC8wDW0p2zKQ7X3lHIdsmKiA==")
	}
	// 获取配置参数，每秒采集是件触发的次数
	tickerTriggers := viper.GetInt64(tickerTriggersKey)

	// 根据每秒的采集触发次数，计算每次的时间间隔
	intervalDuration := time.Duration(1000/tickerTriggers) * time.Millisecond

	// 根据时间间隔创建定时器，调度采集任务的执行
	ticker := time.NewTicker(intervalDuration)

	// mqtt中broker url
	brokerUrl := fmt.Sprintf("mqtt://%s:%d", viper.GetString(mqttServerIPKey),
		viper.GetInt(mqttServerPortKey))

	// 初始化mqtt客户端的相关参数配置
	mqttConfig := mqttclient.MqttConnectionConfig{
		Broker:     brokerUrl,
		ClientID:   viper.GetString(deviceNameKey),
		WillEnable: viper.GetBool(mqttWillEnableKey),
		Qos:        byte(viper.GetInt(mqttQosKey)),
		Retained:   viper.GetBool(mqttRetainedKey),
	}

	// 根据配置信息创建mqtt的客户端连接对象
	var err error
	mqClient, err = mqttclient.NewMqttClient(mqttConfig)
	if nil == mqClient {
		logrus.Error("连接MQTT服务失败:", err.Error())
		return
	} else {
		logrus.Info("连接到MQTT服务")
	}

	// 读取配置，是否开启数据存储(开启:发送到MQTT的数据会写入到influxdb数据库中存储)
	enableStorage := viper.GetBool(deviceStorageEnableKey)
	if enableStorage {
		logrus.Info("开启数据同步存储")
		// influxdb的url
		influxUrl := fmt.Sprintf("http://%s:%d", viper.GetString(influxdbServerIPKey), viper.GetInt(influxdbServerPortKey))

		// influxdb的token,用于登录时的身份验证
		influxToken := viper.GetString(influxdbTokenKey)

		// influxdb异步写入接口，批量写入batchsize数据量时会触发数据写入磁盘
		influxBatchSize := viper.GetUint(influxdbBatchSizeKey)

		// influxdb的客户端对象
		storageClient = influxdb2.NewClientWithOptions(influxUrl, influxToken, influxdb2.DefaultOptions().SetBatchSize(influxBatchSize))

		if nil == storageClient {
			mqClient.Close()
			logrus.Error("连接InfluxDB数据存储服务失败,程序即将退出")
			return
		} else {
			logrus.Info("连接influxdb数据存储服务")
			influxOrg := viper.GetString(influxdbOrgKey)
			influxBucket := viper.GetString(influxdbBucketKey)
			// 根据配置的org和bucket参数，获取异步写入API接口，后续数据写入时使用该接口写入
			storageWriteAPI = storageClient.WriteAPI(influxOrg, influxBucket)
			storageWriteAPI.SetWriteFailedCallback(onInfluxWriteFailed)
		}
	}

	// 清理数据库资源
	defer func() {
		if storageWriteAPI != nil {
			logrus.Info("influxdb flush数据")
			// 把缓冲区的数据写入数据库
			storageWriteAPI.Flush()
		}
		if storageClient != nil {
			logrus.Info("关闭influxdb client连接")
			// 关闭数据库连接
			storageClient.Close()
		}
	}()

	// 关闭mqtt连接
	defer func() {
		if mqClient != nil {
			logrus.Info("关闭Mqtt Client连接")
			mqClient.Close()
		}
	}()

	groupFunc.Go(func() error {
		fmt.Println("启动模拟数据生成协程")
		err := generateAndSendSensorDatas(groupCtx, ticker.C)
		return err
	})

	fmt.Println("程序正在运行中...")
	logrus.Info("程序启动成功，正在运行中...")
	groupFunc.Wait()
	logrus.Info("程序即将退出,清理资源")
}
